package org.apache.flink.runtime.rpc.pekko;

import com.source.flinkrpc.CustomGateway;
import com.source.flinkrpc.CustomRpcEndpoint;
import org.apache.flink.runtime.rpc.pekko.PekkoRpcService;
import org.apache.flink.runtime.rpc.pekko.PekkoRpcServiceConfiguration;
import org.apache.pekko.actor.ActorSystem;


public class FlinkRpcDemo {
    public static void main(String[] args) throws Exception{
         // TODO_MA 注释： 第一步： 启动 RPC 服务
        /** Flink内部可以使用PekkoUtils来创建 ActorSystem*/
        ActorSystem defaultActorSystem = PekkoUtils.createDefaultActorSystem();
        /**
         * 创建RpcService实现对象PekkoRpcService
         *
         */
        PekkoRpcService pekkoRpcService = new PekkoRpcService(defaultActorSystem,
                PekkoRpcServiceConfiguration.defaultConfiguration());
        // TODO_MA 注释： 第二步： 创建 RpcEndpoint 实例，启动 RPC 服务
        CustomRpcEndpoint customRpcEndpoint = new CustomRpcEndpoint(pekkoRpcService);
        customRpcEndpoint.start();

        CustomGateway selfGateway =  customRpcEndpoint.getSelfGateway(CustomGateway.class);
        String getNowResult1 = selfGateway.getName();
        System.out.println(getNowResult1);
         // TODO_MA 注释： 第四步： 通过 RpcEndpoint 地址获得代理
        CustomGateway getNowGateway = pekkoRpcService.connect(customRpcEndpoint.getAddress(), CustomGateway.class).get();
        String getNowResult2 = getNowGateway.getName();
        System.out.println(getNowResult2);
    }
}
